package com.platform.apex.output;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.Date;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;

import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.DefaultInputPort;
import com.datatorrent.common.util.BaseOperator;
import com.platform.apex.cma.db.AllinOneData;
import com.platform.apex.cma.db.MySqlManager;
import com.platform.apex.cma.tools.ConvertUtil;
import com.platform.apex.input.MyJdbcStore;
import com.platform.apex.util.MetaDataUtil;

public class MySqlOutputOperator extends BaseOperator {

	protected transient MyJdbcStore store;
	private String databaseUrl;
	private String userName;
	private String password;
	private String databaseDriver;
	private String tableName;
	protected transient Queue<Object> tupleBatch;

	@Override
	public void setup(OperatorContext context) {
		// TODO Auto-generated method stub
		super.setup(context);
		
		tupleBatch = new ArrayBlockingQueue<Object>(100);
		databaseUrl = databaseUrl+"?useUnicode=true&characterEncoding=UTF-8";
		store = new MyJdbcStore(databaseUrl, userName, password, databaseDriver);
		createTable();
//		MetaDataUtil.initDBConnector(connectionStr, jdbcDriver, userName, password);
//		try {
//			MySqlManager.init(MetaDataUtil.getMetadataConnector());
//		} catch (Exception e) {
//			// TODO Auto-generated catch block
//			e.printStackTrace();
//		}
	}

	public final transient DefaultInputPort<Object> input = new DefaultInputPort<Object>() {
		@Override
		@SuppressWarnings("UseOfSystemOutOrSystemErr")
		public void process(Object tuple) {
			tupleBatch.add(tuple);

			if (tupleBatch.size() >= 100) {
				try {
					batchInsert();
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	};

	@Override
	public void beginWindow(long windowId) {
		// TODO Auto-generated method stub
		super.beginWindow(windowId);
	}

	@Override
	public void endWindow() {
		// TODO Auto-generated method stub
		super.endWindow();
		try {
			batchInsert();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	

	public String getDatabaseUrl() {
		return databaseUrl;
	}

	public void setDatabaseUrl(String databaseUrl) {
		this.databaseUrl = databaseUrl;
	}

	public String getUserName() {
		return userName;
	}

	public void setUserName(String userName) {
		this.userName = userName;
	}

	public String getPassword() {
		return password;
	}

	public void setPassword(String password) {
		this.password = password;
	}

	

	public String getDatabaseDriver() {
		return databaseDriver;
	}

	public void setDatabaseDriver(String databaseDriver) {
		this.databaseDriver = databaseDriver;
	}

	public void batchInsert() throws Exception {
		
		List<Map<String,Object>> list = new ArrayList<Map<String,Object>>();
		while (!tupleBatch.isEmpty()) {
			Object tuple = tupleBatch.remove();
			
			Map<String,Object> map = (Map<String,Object>)tuple;
		
			list.add(map);
		}
		
		insert(list);
//		MySqlManager.create(list);
	}

	public String getTableName() {
		return tableName;
	}

	public void setTableName(String tableName) {
		this.tableName = tableName;
	}
	
	public void insert(List<Map<String, Object>> list) {

		Connection connInsert = null;
		PreparedStatement prest = null;
	
			try {
				store.connect();
				connInsert = store.getConnection();

				// 可以得到有多少列
				connInsert.setAutoCommit(false);
				
				String insertSql = "insert into "+tableName+" (adjust_flow,attact_content,attact_qty, back_opinion, back_opt_id, back_opt_name, back_opt_time, back_status, belong_work, close_opinion, "
						+ " close_opt_id, close_opt_name, close_opt_time, create_dept_id, create_dept_name, create_opt_id, create_opt_name,create_opt_time, data_resource,"
						+ " data_type, deal_opinion, deal_opt_id, deal_opt_name, deal_opt_time, deal_part_id, deal_part_name, deal_result, deal_start_time, deal_status, deal_time,"
						+ " deal_view, detail_id, dispatch_opt_id, dispatch_opt_name, flow_id, is_close, is_valid, is_warn, limit_time, next_dept, next_person, node_id, opt_id, opt_name, "
						+ " opt_order, opt_time, over_status, pre_tran_id, receive_opinion, receive_opt_id, receive_opt_name, receive_opt_time, receive_status, remark, step_id, supervise_id,"
						+ " supervise_level, supervise_opinion, supervise_qty, supervise_title, supervise_type, timely_status, timely_time, tran_id, tran_title, urge_id, urge_opinion, urge_qty, "
						+ " urge_title, warn_content,warn_id, warn_qty, warn_sort, warn_title, warn_type, wish_opt_time, work_id, work_status)"
						+ " VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,? )";
				prest = connInsert.prepareStatement(insertSql, ResultSet.TYPE_SCROLL_SENSITIVE,
						ResultSet.CONCUR_READ_ONLY);
				
				for(Map<String, Object> map : list){
					int count = 0;
					Object[] objects = new Object[map.size()];
					for (Entry<String, Object> en:map.entrySet()) {
						objects[count] = en.getValue();
						if(objects[count]==null){
							prest.setString(count + 1, null);
						}else {
							prest.setString(count + 1, (String) objects[count]);
						}
							
//						if (objects[count] instanceof String) {
//							prest.setString(count + 1, (String) objects[count]);
//						}
//
//						if (objects[count] instanceof Integer) {
//							prest.setInt(count + 1, (int) objects[count]);
//						}
//						
//						if (objects[count] instanceof Boolean) {
//							prest.setBoolean(count + 1, (boolean) objects[count]);
//						}
//
//						if (objects[count] instanceof Date) {
//							prest.setDate(count + 1, (Date) objects[count]);
//						}
//						
//						if (objects[count] instanceof BigDecimal) {
//							prest.setBigDecimal(count + 1, (BigDecimal) objects[count]);
//						}
						count++;
					}
					prest.addBatch();
				}
				prest.executeBatch();
				connInsert.commit();
				connInsert.close();
			} catch (Exception e) {
				// TODO: handle exception
				e.printStackTrace();
			}

	}
	
	/**
	 * adjust_flow attact_content attact_qty back_opinion back_opt_id back_opt_name back_opt_time back_status belong_work close_opinion
	   close_opt_id close_opt_name close_opt_time create_dept_id create_dept_name create_opt_id create_opt_name create_opt_time data_resource
	   data_type deal_opinion deal_opt_id deal_opt_name deal_opt_time deal_part_id deal_part_name deal_result deal_start_time deal_status deal_time
	   deal_view detail_id dispatch_opt_id dispatch_opt_name flow_id is_close is_valid is_warn limit_time next_dept next_person node_id opt_id opt_name
       opt_order opt_time over_status pre_tran_id receive_opinion receive_opt_id receive_opt_name receive_opt_time receive_status remark step_id supervise_id 
       supervise_level supervise_opinion supervise_qty supervise_title supervise_type timely_status timely_time tran_id tran_title urge_id urge_opinion urge_qty
       urge_title warn_content warn_id warn_qty warn_sort warn_title warn_type wish_opt_time work_id work_status
	 * @param tableName
	 */
	public void createTable(){
		
		Connection conn = null;
		   Statement stmt = null;
		   try{
			  store.connect();
		      conn = store.getConnection();
		     
		      stmt = conn.createStatement();

		      
		      String sql = "CREATE TABLE " +tableName+
		                   "(adjust_flow VARCHAR(255),attact_content VARCHAR(255), attact_qty VARCHAR(255), back_opinion VARCHAR(255), back_opt_id VARCHAR(255), back_opt_name VARCHAR(255), back_opt_time VARCHAR(255), back_status VARCHAR(255), belong_work VARCHAR(255),close_opinion VARCHAR(255),"
		                   + " close_opt_id VARCHAR(255), close_opt_name VARCHAR(255), close_opt_time VARCHAR(255), create_dept_id VARCHAR(255), create_dept_name VARCHAR(255), create_opt_id VARCHAR(255), create_opt_name VARCHAR(255), create_opt_time VARCHAR(255), data_resource VARCHAR(255),"
		                   + " data_type VARCHAR(255), deal_opinion VARCHAR(255), deal_opt_id VARCHAR(255), deal_opt_name VARCHAR(255), deal_opt_time VARCHAR(255), deal_part_id VARCHAR(255), deal_part_name VARCHAR(255), deal_result VARCHAR(255), deal_start_time VARCHAR(255), deal_status VARCHAR(255), deal_time VARCHAR(255),"
		                   + " deal_view VARCHAR(255), detail_id VARCHAR(255), dispatch_opt_id VARCHAR(255), dispatch_opt_name VARCHAR(255), flow_id VARCHAR(255), is_close VARCHAR(255), is_valid VARCHAR(255), is_warn VARCHAR(255), limit_time VARCHAR(255), next_dept VARCHAR(255), next_person VARCHAR(255), node_id VARCHAR(255), opt_id VARCHAR(255), opt_name VARCHAR(255),"
		                   + " opt_order VARCHAR(255), opt_time VARCHAR(255), over_status VARCHAR(255), pre_tran_id VARCHAR(255), receive_opinion VARCHAR(255), receive_opt_id VARCHAR(255), receive_opt_name VARCHAR(255), receive_opt_time VARCHAR(255), receive_status VARCHAR(255), remark VARCHAR(255),step_id VARCHAR(255), supervise_id VARCHAR(255),"
		                   + " supervise_level VARCHAR(255), supervise_opinion VARCHAR(255), supervise_qty VARCHAR(255), supervise_title VARCHAR(255), supervise_type VARCHAR(255), timely_status VARCHAR(255), timely_time VARCHAR(255), tran_id VARCHAR(255), tran_title VARCHAR(255), urge_id VARCHAR(255), urge_opinion VARCHAR(255), urge_qty VARCHAR(255),"
		                   + " urge_title VARCHAR(255), warn_content VARCHAR(255), warn_id VARCHAR(255), warn_qty VARCHAR(255), warn_sort VARCHAR(255), warn_title VARCHAR(255), warn_type VARCHAR(255), wish_opt_time VARCHAR(255), work_id VARCHAR(255), work_status VARCHAR(255))";

		      stmt.executeUpdate(sql);
		      System.out.println("Created table in given database...");
		   }catch(Exception e){
		      e.printStackTrace();
		   }finally{
		      try{
		         if(stmt!=null)
		            conn.close();
		      }catch(SQLException se){
		      }
		      try{
		         if(conn!=null)
		            conn.close();
		      }catch(SQLException se){
		         se.printStackTrace();
		      }//end finally try
		   }//end try
	}

}
